import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

object KNNTest {

  import scala.math._
  def getDistance(x: Array[Double], y: Array[Double]): Double = // x y 都是有4个点的数组
    sqrt(x.zip(y).map(elem => pow(elem._1 - elem._2, 2)).sum)

  case class LabeledPoint(label: String, point:Array[Double])

  def main(args: Array[String]): Unit = {
    // 创建SparkContext
    val conf = new SparkConf().setAppName(this.getClass.getCanonicalName.init).setMaster("local[*]")
    val sc = new SparkContext(conf)
    sc.setLogLevel("WARN")

    val fileIris = "F:\\lagou\\lagouhomework\\stage_4_module_2\\4.鸢尾花&KNN算法\\untitled\\data\\Iris.csv"

    val K = 9
    //1.转换为样例类
    val lines = sc.textFile(fileIris)
      .filter(!_.startsWith("Id,"))
      .map{line=>
        val fields = line.split(",")
        if (fields.length == 6) LabeledPoint(fields.last, fields.init.tail.map(_.toDouble))
        else LabeledPoint("", fields.map(_.toDouble))//没有label的将label置为空 也换为LabeledPoint样例类
      }
    //2.分类 分为数据集(已知label的)和测试集（未知名）  将没有label的转换为数组即可 方便做外层循环
    val sampleRDD: RDD[LabeledPoint] = lines.filter(_.label != "")
    val testData: Array[Array[Double]] = lines.filter(_.label == "").map(_.point).collect()

    //小表在外面 相当于join 小表在右边
    testData.foreach(point => {
      sampleRDD.map(labeledpoint => {
        (labeledpoint.label, getDistance(point, labeledpoint.point)) //第一个参数是已知点标签 第二个参数是已知点和未知点距离
      }).sortBy(_._2) //排序
        .take(K) //取前9个
        .map{case (label, _) => label} // 然后第二个参数距离就没啥用了 只取label 做WordCount
        .groupBy(x=>x)
        .mapValues(_.length)
        .foreach(print)
    })

    //随机拷贝一条数据，去除名称，掺在数据文件里做测试用
    //105,6.5,3.0,5.8,2.2,
    /**
     * (Iris-virginica,9)
     */
    // 关闭SparkContext
    sc.stop()
  }
}
